package com.xrui.hbase.impl;

import com.google.common.base.Preconditions;
import com.google.common.collect.Lists;
import com.google.common.collect.Maps;
import com.google.common.collect.Multimap;
import com.google.common.collect.Sets;

import com.xrui.hbase.*;
import com.xrui.hbase.codec.impl.CellDecoderProvider;
import com.xrui.hbase.codec.impl.DefaultCellDecoderFactory;
import com.xrui.hbase.exception.InvalidLayoutException;
import com.xrui.hbase.exception.NoSuchColumnException;
import com.xrui.hbase.filter.RowFilter;
import com.xrui.hbase.schema.TableLayout;
import com.xrui.hbase.schema.spec.CellSpec;
import com.xrui.hbase.schema.spec.ColumnReaderSpec;
import com.xrui.hbase.util.DebugResourceTracker;
import org.apache.hadoop.hbase.client.Get;
import org.apache.hadoop.hbase.client.HTableInterface;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.client.Scan;
import org.apache.hadoop.hbase.filter.Filter;
import org.apache.hadoop.hbase.filter.FilterList;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.util.*;
import java.util.concurrent.atomic.AtomicReference;

public final class HBaseTableReader implements TableReader {

    //region static
    private static final Logger LOG = LoggerFactory.getLogger(HBaseTableReader.class);
    /**
     * HBase Table to read from.
     */
    private final HBaseTable mTable;
    //endregion

    //region Variables
    /**
     * Behavior when a cell decoder cannot be found.
     */
    private final TableReaderBuilder.OnDecoderCacheMiss mOnDecoderCacheMiss;
    /**
     * Map of overridden column read specifications. Null when mCellSpecOverrides is not null.
     */
    private final Map<ColumnName, BoundColumnReaderSpec> mOverrides;
    /**
     * Map of backup column read specifications. Null when mCellSpecOverrides is not null.
     */
    private final Collection<BoundColumnReaderSpec> mAlternatives;
    /**
     * Map of overridden CellSpecs to use when reading. Null when mOverrides is not null.
     */
    private final Map<ColumnName, CellSpec> mCellSpecOverrides;
    private final TableLayout mLayout;
    private final CellDecoderProvider mCellDecoderProvider;
    /**
     * Tracks the state of this TableReader instance.
     */
    private final AtomicReference<State> mState = new AtomicReference<State>(State.UNINITIALIZED);

    //region Constructor
    private HBaseTableReader(
        final HBaseTable table,
        final TableReaderBuilder.OnDecoderCacheMiss onDecoderCacheMiss,
        final Map<ColumnName, ColumnReaderSpec> overrides,
        final Multimap<ColumnName, ColumnReaderSpec> alternatives
    ) throws IOException {
        mTable = table;
        mOnDecoderCacheMiss = onDecoderCacheMiss;

        final TableLayout layout = mTable.getLayout();
        final Set<ColumnName> layoutColumns = layout.getColumns();
        final Map<ColumnName, BoundColumnReaderSpec> boundOverrides = Maps.newHashMap();
        for (Map.Entry<ColumnName, ColumnReaderSpec> override
            : overrides.entrySet()) {
            final ColumnName column = override.getKey();
            if (!layoutColumns.contains(column)
                && !layoutColumns.contains(ColumnName.create(column.getFamily()))) {
                throw new NoSuchColumnException(String.format(
                    "TableLayout: %s does not contain column: %s", layout, column));
            } else {
                boundOverrides.put(column,
                    BoundColumnReaderSpec.create(override.getValue(), column));
            }
        }
        mOverrides = boundOverrides;
        final Collection<BoundColumnReaderSpec> boundAlternatives = Sets.newHashSet();
        for (Map.Entry<ColumnName, ColumnReaderSpec> altsEntry
            : alternatives.entries()) {
            final ColumnName column = altsEntry.getKey();
            if (!layoutColumns.contains(column)
                && !layoutColumns.contains(ColumnName.create(column.getFamily()))) {
                throw new NoSuchColumnException(String.format(
                    "TableLayout: %s does not contain column: %s", layout, column));
            } else {
                boundAlternatives.add(
                    BoundColumnReaderSpec.create(altsEntry.getValue(), altsEntry.getKey()));
            }
        }
        mAlternatives = boundAlternatives;
        mCellSpecOverrides = null;

        mLayout = layout;

        //final CellDecoderProvider provider;
        //if (null != mCellSpecOverrides) {
        //    provider = CellDecoderProvider.create(
        //        layout,
        //        mTable.getHBase().getSchemaTable(),
        //        SpecificCellDecoderFactory.create(),
        //        mCellSpecOverrides);
        //} else {
        mCellDecoderProvider = CellDecoderProvider.create(
            layout,
            mOverrides,
            mAlternatives,
            mOnDecoderCacheMiss);
        //}

        Preconditions.checkState(mLayout != null,
            "TableReader for table: %s failed to initialize.", mTable.getURI());

        // Retain the table only when everything succeeds.
        mTable.retain();
        final State oldState = mState.getAndSet(State.OPEN);
        Preconditions.checkState(oldState == State.UNINITIALIZED,
            "Cannot open TableReader instance in state %s.", oldState);
        DebugResourceTracker.get().registerResource(this);
    }

    public static HBaseTableReader create(
        final HBaseTable table
    ) throws IOException {
        return HBaseTableReaderBuilder.create(table).buildAndOpen();
    }

    //endregion

    //region Public Methods
    public static HBaseTableReader createWithOptions(
        final HBaseTable table,
        final TableReaderBuilder.OnDecoderCacheMiss onDecoderCacheMiss,
        final Map<ColumnName, ColumnReaderSpec> overrides,
        final Multimap<ColumnName, ColumnReaderSpec> alternatives
    ) throws IOException {
        return new HBaseTableReader(table, onDecoderCacheMiss, overrides, alternatives);
    }

    //endregion

    /**
     * Validate a data request against a table layout.
     *
     * @param dataRequest A DataRequest.
     * @param layout      the TableLayout of the table against which to validate the data request.
     */
    private static void validateRequestAgainstLayout(DataRequest dataRequest, TableLayout layout) {
        // TODO(SCHEMA-263): This could be made more efficient if the layout and/or validator were
        // cached.
        DataRequestValidator.validatorForLayout(layout).validate(dataRequest);
    }

    private static Result doHBaseGet(final HBaseTable table, Get get) throws IOException {
        final HTableInterface htable = table.openHTableConnection();
        try {
            LOG.debug("Sending HBase Get: {}", get);
            return htable.get(get);
        } finally {
            htable.close();
        }
    }

    /**
     * Sends a batch of HBase Get requests.
     *
     * @param get HBase Get requests.
     * @return the HBase Results.
     * @throws IOException on I/O error.
     */
    private static Result[] doHBaseGet(final HBaseTable table, List<Get> get) throws IOException {
        final HTableInterface htable = table.openHTableConnection();
        try {
            LOG.debug("Sending bulk HBase Get: {}", get);
            return htable.get(get);
        } finally {
            htable.close();
        }
    }

    /**
     * Creates a list of hbase Gets for a set of entityIds.
     *
     * @param entityIds           The set of entityIds to collect.
     * @param tableLayout         The table layout specifying constraints on what data to return for a row.
     * @param hbaseRequestAdapter The HBaseDataRequestAdapter.
     * @return A list of hbase Gets-- one for each entity id.
     * @throws IOException If there is an error.
     */
    private static List<Get> makeGetList(List<EntityId> entityIds, TableLayout tableLayout,
                                         HBaseDataRequestAdapter hbaseRequestAdapter)
        throws IOException {
        List<Get> hbaseGetList = new ArrayList<Get>(entityIds.size());
        try {
            for (EntityId entityId : entityIds) {
                hbaseGetList.add(hbaseRequestAdapter.toGet(entityId, tableLayout));
            }
            return hbaseGetList;
        } catch (InvalidLayoutException ile) {
            // The table layout should never be invalid at this point, since we got it from a valid
            // opened table.  If it is, there's something seriously wrong.
            throw new InternalError(ile);
        }
    }

    /**
     * Parses an array of hbase Results, returned from a bulk create, to a List of
     * RowData.
     *
     * @param results     The results to parse.
     * @param entityIds   The matching set of EntityIds.
     * @param dataRequest The DataRequest.
     * @return The list of RowData returned by these results.
     * @throws IOException If there is an error.
     */
    private static List<RowData> parseResults(
        final HBaseTable table,
        final Result[] results,
        final List<EntityId> entityIds,
        final DataRequest dataRequest,
        final TableLayout layout
    ) throws IOException {
        List<RowData> rowDataList = new ArrayList<RowData>(results.length);

        for (int i = 0; i < results.length; i++) {
            Result result = results[i];
            EntityId entityId = entityIds.get(i);

            final HBaseRowData rowData = (null == result)
                ? null
                : new HBaseRowData(table, dataRequest, entityId, result,
                CellDecoderProvider.create(layout,
                    layout.getSchemaResolver(),
                    DefaultCellDecoderFactory.get(),
                    Maps.newHashMap()));
            rowDataList.add(rowData);
        }
        return rowDataList;
    }

    //endregion

    //region APIs
    @Override
    public RowData get(EntityId entityId, DataRequest dataRequest)
        throws IOException {
        final State state = mState.get();
        Preconditions.checkState(state == State.OPEN,
            "Cannot create row from TableReader instance %s in state %s.", this, state);

        // Make sure the request validates against the layout of the table.
        validateRequestAgainstLayout(dataRequest, mLayout);

        // Construct an HBase Get to send to the HTable.
        HBaseDataRequestAdapter hbaseRequestAdapter =
            new HBaseDataRequestAdapter(dataRequest);
        Get hbaseGet;
        try {
            hbaseGet = hbaseRequestAdapter.toGet(entityId, mLayout);
        } catch (InvalidLayoutException e) {
            // The table layout should never be invalid at this point, since we got it from a valid
            // opened table.  If it is, there's something seriously wrong.
            throw new InternalError(e);
        }
        // Send the HTable Get.
        final Result result = hbaseGet.hasFamilies() ? doHBaseGet(mTable, hbaseGet) : new Result();

        // Parse the result.
        return new HBaseRowData(
            mTable, dataRequest, entityId, result, CellDecoderProvider.create(mLayout,
            mLayout.getSchemaResolver(),
            DefaultCellDecoderFactory.get(),
            Maps.newHashMap()));
    }

    @Override
    public List<RowData> bulkGet(List<EntityId> entityIds, DataRequest dataRequest)
        throws IOException {

        final State state = mState.get();
        Preconditions.checkState(state == State.OPEN,
            "Cannot create rows from TableReader instance %s in state %s.", this, state);

        // Bulk gets have some overhead associated with them,
        // so delegate work to create(EntityId, DataRequest) if possible.
        if (entityIds.size() == 1) {
            return Collections.singletonList(this.get(entityIds.get(0), dataRequest));
        }

        validateRequestAgainstLayout(dataRequest, mLayout);
        final HBaseDataRequestAdapter hbaseRequestAdapter =
            new HBaseDataRequestAdapter(dataRequest);

        // Construct a list of hbase Gets to send to the HTable.
        final List<Get> hbaseGetList = makeGetList(entityIds, mLayout, hbaseRequestAdapter);

        // Send the HTable Gets.
        final Result[] results = doHBaseGet(mTable, hbaseGetList);
        Preconditions.checkState(entityIds.size() == results.length);

        // Parse the results.  If a Result is null, then the corresponding RowData should also
        // be null.  This indicates that there was an error retrieving this row.
        return parseResults(mTable, results, entityIds, dataRequest, mLayout);

    }

    @Override
    public RowScanner getScanner(DataRequest dataRequest)
        throws IOException {
        return getScanner(dataRequest, new ScannerOptions());
    }

    @Override
    public RowScanner getScanner(DataRequest dataRequest, ScannerOptions scannerOptions)
        throws IOException {
        final State state = mState.get();
        Preconditions.checkState(state == State.OPEN,
            "Cannot create scanner from TableReader instance %s in state %s.", this, state);

        try {
            EntityId startRow = scannerOptions.getStartRow();
            EntityId stopRow = scannerOptions.getStopRow();
            RowFilter rowFilter = scannerOptions.getRowFilter();
            HBaseScanOptions scanOptions = scannerOptions.getHBaseScanOptions();

            final CellDecoderProvider cellDecoderProvider = mCellDecoderProvider;
            final HBaseDataRequestAdapter dataRequestAdapter =
                new HBaseDataRequestAdapter(dataRequest);
            final TableLayout tableLayout = mLayout;
            validateRequestAgainstLayout(dataRequest, tableLayout);
            final Scan scan = dataRequestAdapter.toScan(tableLayout, scanOptions);

            if (null != startRow) {
                scan.setStartRow(startRow.getHBaseRowKey());
            }
            if (null != stopRow) {
                scan.setStopRow(stopRow.getHBaseRowKey());
            }
            scan.setCaching(scannerOptions.getRowCaching());

            if (null != rowFilter) {
                Filter columFilters = scan.getFilter();

                List<Filter> hbaseFilters = Lists.newArrayList();
                hbaseFilters.add(columFilters);
                hbaseFilters.add(rowFilter.toHBaseFilter());

                scan.setFilter(new FilterList(FilterList.Operator.MUST_PASS_ALL, hbaseFilters));
            }

            return new HBaseRowScanner(new HBaseRowScanner.Options()
                .withDataRequest(dataRequest)
                .withTable(mTable)
                .withScan(scan)
                .withCellDecoderProvider(mCellDecoderProvider)
                .withReopenScannerOnTimeout(scannerOptions.getReopenScannerOnTimeout()));
        } catch (InvalidLayoutException e) {
            // The table layout should never be invalid at this point, since we got it from a valid
            // opened table.  If it is, there's something seriously wrong.
            throw new InternalError(e);
        }
    }

    @Override
    public void close() throws IOException {
        final State oldState = mState.getAndSet(State.CLOSED);
        Preconditions.checkState(oldState == State.OPEN,
            "Cannot close TableReader instance %s in state %s.", this, oldState);
        DebugResourceTracker.get().unregisterResource(this);
        mTable.release();
    }

    //endregion

    //region Private Methods

    /**
     * States of a table reader instance.
     */
    private static enum State {
        UNINITIALIZED,
        OPEN,
        CLOSED
    }

    //endregion
}
